package event;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Scanner;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class BasicInit {
	public static final String INFO = "Info";
	public static final String NAME = "Name";
	public static final String TOTAL = "Total";
	public static final String NUMBER = "Number";
	public static final String NODE = "Node";
	private PageConfig nodeConfig = null;
	public static void main(String [] args)
	{
		BasicInit basicLoad = new BasicInit();
		for(String arg : args)
		{
			basicLoad.loadBasic(arg);
		}
		basicLoad.createNode();
	}
	public BasicInit()
	{
		nodeConfig = new PageConfig();
	}
	//load the data for attributes into database
	private void loadBasic(String fileName) {
		int i = 0;
		Scanner scanner = null;
		try {
			scanner = new Scanner(new FileInputStream(fileName));
			HTableDescriptor htd = new HTableDescriptor(fileName);
			htd.addFamily(new HColumnDescriptor(INFO));
			Configuration config = HBaseConfiguration.create();
			HBaseAdmin admin = new HBaseAdmin(config);
			admin.createTable(htd);
			byte [] tablename = htd.getName();
			HTable table = new HTable(config, tablename);
			byte [] databytes = Bytes.toBytes(INFO);
			while (scanner.hasNextLine()){
				byte [] row = Bytes.toBytes(i);
				Put p1 = new Put(row);
				String value = scanner.nextLine();
				p1.add(databytes, Bytes.toBytes(NAME), Bytes.toBytes(value));
				table.put(p1);
				i++;
			}
			byte [] row = Bytes.toBytes(TOTAL);
			Put p1 = new Put(row);
			p1.add(databytes, Bytes.toBytes(NUMBER), Bytes.toBytes(i));
			table.put(p1);
			table.close();
			admin.close();
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	//create the database table to store the links between pages
	private void createNode()
	{
		try {
			HTableDescriptor htd = new HTableDescriptor(NODE);
			Set<String> linkSet =  new HashSet<String>();
			List<EventNode> eventNodeList = nodeConfig.getEventNodeList();
			for(int i = 0; i < eventNodeList.size(); i++)
			{
				EventNode eventNode = eventNodeList.get(i);
				List<Link> linkList = eventNode.getLinkList();
				for(int j = 0; j < linkList.size(); j++)
				{
					Link link = linkList.get(j);
					String linkName = link.getLinkName();
					if(!linkSet.contains(linkName))
					{
						linkSet.add(linkName);
					}
				}
			}
			for(String link : linkSet)
			{
				htd.addFamily(new HColumnDescriptor(link));
			}
			Configuration config = HBaseConfiguration.create();
			HBaseAdmin admin = new HBaseAdmin(config);
			admin.createTable(htd);
			admin.close();
			
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
